import com.o2o.utils.Iargs
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/6/5 16:23
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object CateDataJoin {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .config("spark.debug.maxToStringFields", "2000")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.caseSensitive", "true")
      .config("es.nodes", "192.168.1.29")
      .config("es.port", "9200")
      .config("cluster.name","O2OElastic")
      .config("es.net.http.auth.user", "elastic")
      .config("es.net.http.auth.pass", "changeme")
//      .master("local[*]")
      .getOrCreate()

    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", Iargs.OBSACCESS)
    sc.hadoopConfiguration.set("fs.s3a.secret.key", Iargs.OBSSECRET)
    sc.hadoopConfiguration.set("fs.s3a.endpoint", Iargs.OBSENDPOINT)
    sc.setLogLevel("WARN")
//    spark.read.orc("s3a://dws-data/g_data/2020/10/meituan/").printSchema()
//    println(spark.readsv("s3a://dws-data/backData_month/O-O-DB/dbadmin/o2o_anchor_APP_2020/2020-10-12-19/").count())

val frame: DataFrame = spark.read.orc("s3a://dws-data/g_data/2020/10/meituan/")
  .where("province = '广东省'").drop("address", "administrative_region", "aedzId", "city", "city_grade", "city_origin", "district", "district_origin", "economic_division", "if_city", "if_district", "if_state_level_new_areas", "latitude", "longitude", "poor_counties", "province", "regional_ID", "registration_institution", "rural_demonstration_counties", "rural_ecommerce", "the_belt_and_road_city", "the_belt_and_road_province", "the_yangtze_river_economic_zone_city", "the_yangtze_river_economic_zone_province", "town", "urban_agglomerations"
)


frame.registerTempTable("source")



spark.read.json("s3a://o2o-dataproces-group/yang_songjian/product/meituan/2020/7/resultAddress/*").registerTempTable("addr")

val guangdongDF = spark.sql(
  """
    |select
    |a.*,
    |b.*
    |from
    |source a
    |left join
    |addr b
    |on a.shopId = b.shopId
  """.stripMargin).selectExpr("address", "administrative_region", "aedzId", "categoryId", "categoryName", "city", "city_grade", "city_origin", "county", "district", "district_origin", "dpShopId",
  "economic_division", "emotionalKeywords", "evaluates", "firstCategoryId", "flavors", "food_type", "fourthCategoryId", "goodDescription", "goodRatePercentage", "good_id", "if_city",
  "if_district", "if_state_level_new_areas", "images", "is_brand", "is_premium", "latitude", "licencePics", "longitude", "mtWmPoiId", "mtWmPoiIdNew", "opening_hours", "order_lead_time",
  "original_cost", "packing_fee", "phone", "platformId", "platformName", "poor_counties", "praiseNum", "priceText", "promotion_info", "province", "regional_ID", "registration_institution",
  "rootCategoryId", "rootCategoryName", "rural_demonstration_counties", "rural_ecommerce", "secondCategoryId", "shopCommentCount", "shopDescription", "shopId", "shopImages", "shopImg",
  "shopName", "shopSellCount", "shopUrl", "shopUuid", "shop_open", "shop_open_day", "star", "street", "the_belt_and_road_city", "the_belt_and_road_province", "the_yangtze_river_economic_zone_city",
  "the_yangtze_river_economic_zone_province", "thirdCategoryId", "third_category_name", "timeStamp", "title", "town", "urban_agglomerations", "sellCount", "salesAmount")

val otherDF: Dataset[Row] = spark.read.orc("s3a://dws-data/g_data/2020/10/meituan/")
  .where("province != '广东省'")

    val frame1: DataFrame = otherDF.selectExpr("address", "administrative_region", "aedzId", "categoryId", "categoryName", "city", "city_grade", "city_origin", "county", "district", "district_origin", "dpShopId",
      "economic_division", "emotionalKeywords", "evaluates", "firstCategoryId", "flavors", "food_type", "fourthCategoryId", "goodDescription", "goodRatePercentage", "good_id", "if_city",
      "if_district", "if_state_level_new_areas", "images", "is_brand", "is_premium", "latitude", "licencePics", "longitude", "mtWmPoiId", "mtWmPoiIdNew", "opening_hours", "order_lead_time",
      "original_cost", "packing_fee", "phone", "platformId", "platformName", "poor_counties", "praiseNum", "priceText", "promotion_info", "province", "regional_ID", "registration_institution",
      "rootCategoryId", "rootCategoryName", "rural_demonstration_counties", "rural_ecommerce", "secondCategoryId", "shopCommentCount", "shopDescription", "shopId", "shopImages", "shopImg",
      "shopName", "shopSellCount", "shopUrl", "shopUuid", "shop_open", "shop_open_day", "star", "street", "the_belt_and_road_city", "the_belt_and_road_province", "the_yangtze_river_economic_zone_city",
      "the_yangtze_river_economic_zone_province", "thirdCategoryId", "third_category_name", "timeStamp", "title", "town", "urban_agglomerations", "sellCount", "salesAmount")


    frame1.union(guangdongDF).where("shopId not in ('24ad4b2af5c459b3','04649e066fc7d147','70444d3d7652f165','d687dcf9801e71f6','64945de50d94f62d','7c62027435be5dfd','bb7765181620428f','edb2b43f0741c793','ad8d72578054b1da')")
      .write.orc("s3a://dws-data/g_data/2020/10/meituan_new/")

sc.stop()
}
}
